package cn.gxufe.reliable.message.consumer.kafka;

import cn.gxufe.reliable.message.api.consumer.ProcessMqMessage;
import cn.gxufe.reliable.message.api.mq.ByteRMSConsumer;
import cn.gxufe.reliable.message.common.Message;
import com.google.gson.Gson;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author yingjie.rong
 **/
@Component
public class KafkaRMSConsumer implements ByteRMSConsumer {
    private Gson gson = new Gson();
    @Autowired
    private ProcessMqMessage handleMqMessage;

    @Override
    @KafkaListener(topics = "rmsKafkaQueue",groupId = "rms_kafka_queue")
    public void listen(String data) {
        Message message = gson.fromJson(data, Message.class);
        handleMqMessage.handle(message);
    }
}
